package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDao;
import com.atguigu.userprofile.util.MyClickhouseUtil;
import com.atguigu.userprofile.util.MySQLUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitMapApp {
    public static void main(String[] args) {
        //虽然用不到，但是为了能够在spark环境中运行，因此还要创建spark对应的执行环境
        SparkConf sparkConf = new SparkConf().setAppName("TaskBitMapApp");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();
        //获取传参 busiDate
        String busiDate = args[1];

        //从mysql中获取所有已启用的标签
        List<TagInfo> tagInfoWithOnList = TagInfoDao.getTagInfoWithOn();

        //根据标签值类型来判断，数据应该导入到哪张表中
        //创建四个集合根据标签值不同的类型，将tagInfo放入对应类型的集合中
        ArrayList<TagInfo> stringTagInfo = new ArrayList<>();
        ArrayList<TagInfo> longTagInfo = new ArrayList<>();
        ArrayList<TagInfo> decimalTagInfo = new ArrayList<>();
        ArrayList<TagInfo> dateTagInfo = new ArrayList<>();

        for (TagInfo tagInfo : tagInfoWithOnList) {
            if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagInfo.getTagValueType())){
                longTagInfo.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagInfo.getTagValueType())){
                decimalTagInfo.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())){
                stringTagInfo.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())){
                dateTagInfo.add(tagInfo);
            }
        }

        //插入语句
        insertBitMapSQL(stringTagInfo, busiDate, "user_tag_value_string");
        insertBitMapSQL(longTagInfo, busiDate, "user_tag_value_long");
        insertBitMapSQL(decimalTagInfo, busiDate, "user_tag_value_decimal");
        insertBitMapSQL(dateTagInfo, busiDate, "user_tag_value_date");
    }

    public static void insertBitMapSQL(List<TagInfo> list,String busiDate,String insertTableName) {
        /**
         * 插入语句
         * insert into table user_tag_value_string
         * select tg.1, tg.2, groupBitmapState(uid) as us
         * from (
         *       select uid,
         *              arrayJoin([('gender', gender), ('agegroup', agegroup), ('favor', favor)]) as tg
         *       from user_tag_merge
         *          )
         * group by tg.1, tg.2
         */

        /**
         * 查询语句
         * select tg.1, tg.2, groupBitmapState(uid) as us
         *          * from (
         *          *       select uid,
         *          *              arrayJoin([('gender', gender), ('agegroup', agegroup), ('favor', favor)]) as tg
         *          *       from user_tag_merge
         *          *          )
         *          * group by tg.1, tg.2
         */

        if (list.size() > 0) {
            //编辑Tuple2元组中的语句
            List<String> tagCodeWithTuple2List = list.stream().map(tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "', " + tagInfo.getTagCode().toLowerCase() + ")").collect(Collectors.toList());
            String tagCodeTuple2SQL = StringUtils.join(tagCodeWithTuple2List, ",");

            //表名
            String tableName = "up_tag_merge_" + busiDate.replace("-", "");

            String selectSQL = "select tg.1, tg.2, groupBitmapState(uid) as us,'"+busiDate +
                    "'    from (\n" +
                    "      select cast(uid as UInt64) as uid,\n" +
                    "      arrayJoin([" + tagCodeTuple2SQL + "]) as tg\n" +
                    "      from " + tableName + "\n" +
                    "         )\n" +
                    "      group by tg.1, tg.2 ";

            //为了防止重跑的时候有脏数据，提前把当天跑失败的数据删掉，然后在重新跑
            String deleteSQL = " alter table " + insertTableName + " delete where dt='" + busiDate + "'";

            String insertSQL = "  insert into table " + insertTableName + " " + selectSQL;

            System.out.println(selectSQL);

            MyClickhouseUtil.executeSql(deleteSQL);
            MyClickhouseUtil.executeSql(insertSQL);
        }
    }
}
